package com.ssm.web.config.mq;

import com.ssm.common.constant.Constants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 消息发送确认
 * <p>
 * ConfirmCallback 只确认消息是否正确到达 Exchange 中
 * ReturnCallback 消息没有正确到达队列时触发回调，如果正确到达队列不执行
 * <p>
 * 1. 如果消息没有到 exchange，则 confirm 回调，ack=false
 * 2. 如果消息到达 exchange，则 confirm 回调，ack=true
 * 3. exchange 到 queue成功，则不回调 return
 * 4. exchange 到 queue失败，则回调 return
 */
@Slf4j
@Configuration(proxyBeanMethods = false)
public class RabbitConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    private final RabbitTemplate rabbitTemplate;

    private Map<String, Map<String, Object>> queueMap = new ConcurrentHashMap<>();

    public RabbitConfig(RabbitTemplate rabbitTemplate) {
        rabbitTemplate.setConfirmCallback(this::confirm);
        rabbitTemplate.setReturnCallback(this::returnedMessage);
        this.rabbitTemplate = rabbitTemplate;
    }

    /**
     * 发送消息中间件数据
     */
    public void sendMQData(Object mqData) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        queueMap.put(correlationData.getId(), this.createMap(Constants.MQ_DATA_QUEUE, mqData));
        rabbitTemplate.convertAndSend(Constants.MQ_DATA_QUEUE, mqData, correlationData);
    }

    /**
     * 发送消息回调
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            // 发送成功
            if (ObjectUtils.isEmpty(correlationData) || StringUtils.isEmpty(correlationData.getId())) {
                return;
            }
            queueMap.remove(correlationData.getId());
        } else {
            // 重试
            Map<String, Object> map = queueMap.get(correlationData.getId());
            rabbitTemplate.convertAndSend(map.get("queueName").toString(), map.get("mqData"), correlationData);
        }
    }

    /**
     * 实现 ReturnCallback
     * 当消息发送出去找不到对应路由队列时，将会把消息退回
     * 如果有任何一个路由队列接收投递消息成功，则不会退回消息
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.error(String.format("消息发送出去找不到对应路由队列,消息使用的交换器 exchange:%s,消息使用的路由键 routing:%s,消息体:%s",
                exchange, routingKey, message.toString()));
    }

    private Map<String, Object> createMap(String queueName, Object mqData) {
        Map<String, Object> map = new HashMap<>(5);
        map.put("queueName", queueName);
        map.put("mqData", mqData);
        return map;
    }
}